package com.systar.activity.process;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

import com.google.common.collect.ArrayListMultimap;
import com.google.common.collect.ListMultimap;
import com.systar.activity.events.IActivityEvent;
import com.systar.activity.monitor.ActivitySource;

public class ActivitySortReduce
{
	// CONCAT Events maps
	private ListMultimap<String, IActivityEvent> eventByType;
	private Map<ActivitySource, IActivityEvent> eventBySource;
	
	public ActivitySortReduce()
	{
		eventByType = ArrayListMultimap.create();
		eventBySource = new HashMap<ActivitySource, IActivityEvent>();
	}

	public synchronized List<IActivityEvent> process(List<IActivityEvent> outputStack)
	{
		System.out.println("Entering SortReduce.process");
		if (outputStack.size() < 2) // no need to concat anyway
		{
			System.out.println("Skipping SortReduce.process");
			return outputStack;
		}
		
		// SORT : sort events by type
		for (IActivityEvent e : outputStack)
		{
			String type = e.getType();
			eventByType.put(type, e);
		}
		outputStack.clear();

		// REDUCE : compact events by source
		for (String type : eventByType.keySet())
		{
			List<IActivityEvent> list = eventByType.get(type);
			if (list.size() < 2) // no need to concat anyway
			{
				outputStack.addAll(list);
			}
			else
			{
				eventBySource.clear();
				// concat by type and by source
				for (IActivityEvent event : list)
				{
					ActivitySource src = event.getSource();
					if (eventBySource.containsKey(src))
						eventBySource.get(src).concat(event);
					else
						eventBySource.put(src, event.clone());
				}
				
				outputStack.addAll(eventBySource.values());
			}
		}
		
		// clear buffers
		eventByType.clear();
		eventBySource.clear();
		
		System.out.println("Leaving SortReduce.process");
		return outputStack;
	}
}
